package com.zyb.dfs.backupnode.server;


import java.io.File;
import java.io.FileOutputStream;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * 将fsimage的checkpoint的组件
 */
public class FsImageCheckpoiner extends Thread {
    public static final Integer CHECKPOINT_INTERVAL = 30 * 1000;

    private BackupNode backupNode;

    private FSNamesystem namesystem;

    private NameNodeRpcClient namenode;

    private String lastFsimageFile = "";

    private long checkpointTime = System.currentTimeMillis();

    /**
     * 上一次checkpoint的文件
     */
    private String lastCheckpointFile = "";

    public FsImageCheckpoiner(BackupNode backupNode, FSNamesystem namesystem, NameNodeRpcClient nameNode) {
        this.backupNode = backupNode;
        this.namesystem = namesystem;
        this.namenode = nameNode;
    }

    @Override
    public void run() {
        System.out.println("fsimage checkpoint定时调度线程启动......");

        while(backupNode.isRunning()) {
            try {
                if(!namesystem.isFinishedRecover()) {
                    System.out.println("当前还没完成元数据恢复，不进行checkpoint......");
                    Thread.sleep(1000);
                    continue;
                }

                if(lastFsimageFile.equals("")) {
                    this.lastFsimageFile = namesystem.getCheckpointFile();
                }

                long now = System.currentTimeMillis();

                if(now - checkpointTime > CHECKPOINT_INTERVAL) {
                    /*if(!namenode.isNamenodeRunning()) {
                        System.out.println("namenode当前无法访问，不执行checkpoint......");
                        continue;
                    }*/

                    // 就可以触发这个checkpoint操作，去把内存里的数据写入磁盘就可以了
                    // 在写数据的这个过程中，你必须是
                    System.out.println("准备执行checkpoint操作，写入fsimage文件......");
                    doCheckpoint();
                    System.out.println("完成checkpoint操作......");
                }

                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 将fsimage持久化到磁盘上去
     *
     * @throws Exception
     */
    public void doCheckpoint() throws Exception {
        FSImage fsimage = namesystem.getFSImageJson();
        removeLastFsimage();
        writeFSImageFile(fsimage);
        uploadFSImageFile(fsimage);
        updateCheckpointTxid(fsimage);
        saveCheckpointInfo(fsimage);
    }

    /**
     *
     * @param fsimage
     */
    private void saveCheckpointInfo(FSImage fsimage) {
        String path = "Users/admin/backupnode/checkpoint-info.meta";

        RandomAccessFile raf = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            File file = new File(path);
            if(file.exists()) {
                file.delete();
            }

            long now = System.currentTimeMillis();
            this.checkpointTime = now;
            long checkpointTxid = fsimage.getMaxTxid();
            ByteBuffer buffer = ByteBuffer.wrap(String.valueOf(now + "_" + checkpointTxid + "_" + lastFsimageFile).getBytes());

            raf = new RandomAccessFile(path, "rw");
            out = new FileOutputStream(raf.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false);

            System.out.println("checkpoint信息持久化到磁盘文件......");
        } catch(Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if(out != null) {
                    out.close();
                }
                if(raf != null) {
                    raf.close();
                }
                if(channel != null) {
                    channel.close();
                }
            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }
    }

    /**
     * 跟新checkpiont
     *
     * @param fsimage
     */
    private void updateCheckpointTxid(FSImage fsimage) {
        namenode.updateCheckpointTxid(fsimage.getMaxTxid());
    }

    /**
     * 删除上一个fsimage磁盘文件
     */
    private void removeLastFsimage() {
        File file = new File(lastCheckpointFile);
        if (file.exists()) {
            file.delete();
        }
    }

    /**
     * 写入最新的fsimage文件
     *
     * @throws Exception
     */
    private void writeFSImageFile(FSImage fsimage) throws Exception {
        RandomAccessFile file = null;
        FileOutputStream out = null;
        FileChannel disLogFileChannel = null;
        ByteBuffer buffer = ByteBuffer.wrap(fsimage.getFsimageJson().getBytes());
        String fsimageFilePath = "/Users/admin/backupnode/fsimage-" + fsimage.getMaxTxid() + ".mate";
        lastCheckpointFile = fsimageFilePath;
        try {
            file = new RandomAccessFile(fsimageFilePath, "rw");
            out = new FileOutputStream(file.getFD());
            disLogFileChannel = out.getChannel();
            disLogFileChannel.write(buffer);
            // 强制把数据刷入到磁盘
            disLogFileChannel.force(false);
        } catch (Exception e) {
            e.getStackTrace();
        } finally {
            if (out != null) {
                out.close();
            }
            if (file != null) {
                file.close();
            }
            if (disLogFileChannel != null) {
                disLogFileChannel.close();
            }
        }
    }

    /**
     * 上传fsimage文件
     *
     * @param fsimage
     */
    private void uploadFSImageFile(FSImage fsimage) {
        FSImageUploader fsImageUploader = new FSImageUploader(fsimage);
        fsImageUploader.start();
    }
}
